<?php
namespace zake\server;

use zake\helper\Console;
use zake\foundation\Process;
use zake\helper\Session;
use zake\server\zadb\ZaListen;
use zake\server\ws\WsConnect;
use zake\server\http\HttpConnect;
use zake\foundation\reactor\BaseLoop;
use zake\foundation\reactor\Reactor;
use zake\foundation\able\reactor\ListenAble;
use zake\foundation\reactor\driver\EventAdapt;
use zake\foundation\reactor\driver\SocketPairChannelAdapt;
use zake\foundation\reactor\trigger\Channel;
use zake\foundation\reactor\driver\SelectAdapt;
use zake\foundation\able\reactor\ChannelAble;
use zake\foundation\reactor\trigger\BeeConnect;

/**
 * Class BaseServer
 * @package zake\server
 */
abstract class BaseServer
{

    /**
     * 子进程数量
     * @var int
     */
    public $num = 2;


    /**
     *
     * @param $processId
     * @param ListenAble $listener
     * @param null $channel
     *
     * @author liu.bin 2020-05-18 16:56
     */
    protected function runDb($processId,ListenAble $listener,$channel=null)
    {
        /**
         * @var ListenAble $listener
         *
         * 初始化loop
         */
        $loop = Reactor::singleton(SelectAdapt::create());
        $protocol = $listener->getProtocol();
        Reactor::addReadListener($listener,function() use($listener,$protocol){

            // 接收 work进程 的链接
            $connect = $listener->acceptConnect();
            if(empty($connect)){
                return;
            }

            // connect 读
            Reactor::addReadListener($connect,function() use($connect,$protocol){
                // read buffer.可触发buffer和close事件
                $connect->readBuffer($protocol,function($buffer) use($connect){
                    if($connect instanceof HandlerAble){
                        $connect->onBufferHandler($buffer);
                    }
                },function()use($connect){
                    if($connect instanceof HandlerAble){
                        $connect->onCloseHandler();
                    }
                });
            });
        });


        Console::success("---- db server [{$processId}] running ----");
        $loop->loop();
    }



    /**
     * 子进程
     *
     * @param int $processId
     * @param BaseLoop $loop
     * @param Channel|null $channel
     *
     * @author liu.bin 2020-04-14 16:45
     */
    protected function runWork($processId,$listenerList,$channel=null)
    {
        /**
         * @var ListenAble $listener
         *
         * 初始化loop
         */
        $loop = Reactor::singleton(EventAdapt::create());

        // channel添加可读事件
        Reactor::addReadListener($channel,function()use($channel){
            $this->workChannel($channel);
        });

        // listen 添加可读事件
        foreach ($listenerList as $listener){
            $listener->setChannel($channel);
            $protocol = $listener->getProtocol();
            Reactor::addReadListener($listener,function() use($listener,$protocol){

                // 惊群的问题
                // 多进程模式下，listenSocket可读时，所有进程会被唤醒，只有一个进程会accept成功
                // 如果返回null，表明该进程被唤醒，但是没有accept成功，需要再次进入reactor监听
                $connect = $listener->acceptConnect();
                if(empty($connect)){
                    return;
                }

                // connect 读
                Reactor::addReadListener($connect,function() use($connect,$protocol){
                    // read buffer.可触发buffer和close事件
                    $connect->readBuffer($protocol,function($buffer) use($connect){
                        if($connect instanceof HandlerAble){
                            $connect->onBufferHandler($buffer);
                        }
                    },function()use($connect){
                        if($connect instanceof HandlerAble){
                            $connect->onCloseHandler();
                        }
                    });
                });
            });
        }



        // 定时器
        Reactor::addTimeListener(10,function(){
            $connectMap    = Reactor::getConnectMap();
            $nowTime        = time();


            foreach ($connectMap as $connect){

                // http 超时3s 断开
                if($connect instanceof HttpConnect){
                    $readTime = $connect->getReadTime();
                    if($nowTime - $readTime >= 3){
                        $connect->close();
                    }
                    continue;
                }

                // webSocket ping 心跳检测
                if($connect instanceof WsConnect){
                    $ping = $connect->getProtocol()->encodePing();
                    $connect->writeBuffer($ping,null,['encode'=>false]);
                    continue;
                }

            }
        });



        Console::success("---- work server [{$processId}] running ----");
        $loop->loop();
    }


    /**
     * todo 子进程 channel 业务
     * @param ChannelAble $channel
     *
     * @author liu.bin 2020-05-15 23:17
     */
    protected function workChannel(ChannelAble $channel)
    {

        $protocol = $channel->getProtocol();

        $channel->readBuffer($protocol,function($buffer){

        },function(){

        });
    }



    /**
     * todo 主进程 channel 业务
     * channel可读时触发
     *
     * @param ChannelAble $channel
     *
     * @author liu.bin 2020-04-14 15:36
     */
    protected function masterChannel(ChannelAble $channel)
    {
        $protocol = $channel->getProtocol();
        $channel->readBuffer($protocol,function($buffer){

        },function(){

        });
    }


    /**
     * todo 集群 channel 业务
     * @param BeeConnect $connect
     *
     * @author liu.bin 2020-05-15 23:51
     */
    protected function clusterChannel(BeeConnect $connect)
    {
        $protocol = $connect->getProtocol();
        $connect->readBuffer($protocol,function($buffer){

        },function(){

        });
    }


    /**
     * 开启循环事件处理
     *
     * @author liu.bin 2020-05-15 17:13
     */
    protected function loop()
    {
        /**
         * @var ListenAble $listener
         * @var ChannelAble $channel
         * 创建channel
         * 创建子进程
         * 处理channel消息
         */


        // 创建table进程，用作数据共享进程
        $zaListener = ZaListen::create(ZaListen::URI);
        $process    = Process::create(
            function($processId,$channel) use($zaListener){
                $this->runDb($processId,$zaListener,$channel);
            },
            $channel
        );
        $process->start();



        // 创建work进程，接收客户端连接
        $channelList        = [];
        $listenerList       = $this->getListenerList();
        for($i = 0; $i< $this->num; $i++){
            $channel    = Channel::create(SocketPairChannelAdapt::create());
            $process    = Process::create(
                function($processId,$channel) use($listenerList){
                    $this->runWork($processId,$listenerList,$channel);
                },
                $channel
            );

            // 开启子进程
            $process->start();
            array_push($channelList,$channel);
        }


        // 初始化loop
        $loop = Reactor::singleton(SelectAdapt::create());


        // 关闭主进程 listen socket
        foreach ($listenerList as $listener){
            $listener->closeResource();
            unset($listener);
        }
        $zaListener->closeResource();
        unset($zaListener);


        // 接收子进程消息，缓存数据
        foreach ($channelList as $channel){
            // todo 定义channel sessionId
            $channel->setSessionId(Session::id(1,1,0));
            Reactor::addReadListener($channel,function()use($channel){
                $this->masterChannel($channel);
            });
        }


        // 服务开启
        $loop->loop();
        Console::error("-----------master server wait-----------");
        Process::wait();
        Console::error("-----------master server exit-----------");
    }


    /**
     * 获取listener列表
     * @return array
     *
     * @author liu.bin 2020-05-06 22:30
     */
    abstract public function getListenerList();

}